package chapter01.part14;

import chapter01.part1.CountWordByLength;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

import static java.util.stream.Collectors.*;

/**
 * 并行流的基本使用
 * 流的出现 使得并行处理数据库变得容易
 * 
 * 1. 获取并行流的方式
 * 2. 并行操作必须能以任意顺序执行 不影响结果
 * 3. 排序会影响并行效率
 * 4. 并行不是万能的 需要大量开销
 * 5. 数据源必须可以被分割
 * 6. 并行流使用线程池 可能会因为 IO 操作阻塞而死
 * 
 * @author insight
 * @since 2021/8/2
 */
public class ParallelStream {
    public static void main(String[] args) {
        // 获取并行流
        // 1. 使用 collection.parallelStream()
        // 2. 将已有的流转换为并行流
        List<String> words = CountWordByLength.getWordList();
        Stream<String> parallelWords = words.parallelStream();
        Stream<String> parallel = words.stream().parallel();
        
        // 并发问题 数据量越大 误差越大
        int len = 10;
        var shortWords = new int[len];
        words.parallelStream().forEach(
                w -> {
                    if (w.length() < len) shortWords[w.length()]++;
                }
        );
        System.out.println(Arrays.toString(shortWords));
        // 再来一次 结果不同
        Arrays.fill(shortWords, 0);
        words.parallelStream().forEach(
                w -> {
                    if (w.length() < len) shortWords[w.length()]++;
                }
        );
        System.out.println(Arrays.toString(shortWords));
        // 解决方法 远离易变状态
        // 按长度分组 然后对每一组分别统计
        Arrays.fill(shortWords, 0);
        Map<Integer, Long> collect = words.parallelStream()
                .filter(s -> s.length() < len)
                .collect(
                        groupingBy(String::length, counting())
                );
        System.out.println(collect);
        
        // 排序和并行效率
        // 有序集合的流 也是有序的 以及 stream().sorted()
        // 此处指定长度排序
        words.sort(Comparator.comparing(String::length));
        Stream<String> sortedStream = words.parallelStream();
        // 有序的流可以被划分并行处理 合并时会按照顺序组装
        System.out.println(sortedStream.map(String::toLowerCase)
                .collect(Collectors.toList()));
        // 通过调用 unordered() 放弃排序需求 可以更高效地并行处理
        // 注意: 经过测试 unordered 对有序集合才有效果 比如 TreeSet
        TreeSet<String> wordSet = new TreeSet<>(words);
        // distinct() 默认保留相同元素的第一个
        // 使用无序可以突破这种约束 保留任意一个
        for (int i = 0; i < 10; i++) {
            System.out.println(wordSet.parallelStream().unordered().distinct().collect(Collectors.toList()));
        }        
        // 同理 limit 也可取出任意 n 个 每次结果不同
        for (int i = 0; i < 10; i++) {
            System.out.println(wordSet.parallelStream().unordered().limit(2).collect(Collectors.toList()));
        }
        
        // 还可以使用并行流来合并 Map 提高效率 相应的 顺序会发生改变
        ConcurrentMap<Integer, List<String>> wordMap =
                words.parallelStream().collect(
                groupingByConcurrent(
                        String::length
                )
        );
        
        // 并行流默认使用 Fork-Join 也可使用不同的池 或使用异步
        ForkJoinPool customPool = new ForkJoinPool();
        customPool.submit(() ->
                words.parallelStream().map(String::toUpperCase).collect(toList())
        );
        CompletableFuture.supplyAsync(() ->
                words.parallelStream().map(String::toUpperCase).collect(toList()),
                customPool
        );
        
        // 最后要注意的是 随机数流
        // 可以使用 Random.ints() 获取基本类型流
        IntStream intStream = new Random().ints().limit(10);
        // 但是不要使用这种流作为并行流的起点 因为他们不可分割
        // 使用可分割的随机数流
        IntStream splittable = new SplittableRandom().ints().limit(10).parallel();
    }
    
}